package com.boarsoft.boar.message.action;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Vector;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import com.boarsoft.bean.ReplyInfo;
import com.boarsoft.boar.common.Constants;
import com.boarsoft.boar.message.adapter.MessageBrokerAdapter;
import com.boarsoft.boar.message.biz.TargetQueueBiz;
import com.boarsoft.boar.message.entity.TargetQueueEntity;
import com.boarsoft.boar.message.service.QueueMonitor;
import com.boarsoft.boar.service.FileService;
import com.boarsoft.common.Authorized;
import com.boarsoft.common.Util;
import com.boarsoft.common.dao.PagedResult;
import com.boarsoft.message.bean.MessageQueueStat;
import com.boarsoft.message.bean.TargetQueueInfo;

@RestController
@RequestMapping("/target/queue")
public class TargetQueueAction {
	private static final Logger log = LoggerFactory.getLogger(TargetQueueAction.class);

	@Autowired
	private MessageBrokerAdapter brokerAdapter;

	@Autowired
	private TargetQueueBiz targetQueueBiz;

	@Autowired(required = false)
	protected FileService fileService;

	@Autowired
	@Qualifier("targetQueueMonitor")
	private QueueMonitor targetQueueMonitor;

	@RequestMapping("/stat.do")
	public ReplyInfo<Object> stat(String id) {
		String broker=targetQueueBiz.get(id).getBroker();
		MessageQueueStat so = targetQueueMonitor.getStat(broker,id);
		return new ReplyInfo<Object>(true, so);
	}

	/**
	 * 从数据库查询消息队列的信息
	 */
	@SuppressWarnings("unchecked")
	@RequestMapping("/list.do")
	@Authorized(code = "message.target.queue.list")
	public ReplyInfo<Object> list(String catalog, String key, String orderBy, int pageNo, int pageSize) {
		Map<String, TargetQueueEntity> qeMap = new HashMap<String, TargetQueueEntity>();
		//
		PagedResult<TargetQueueEntity> pr = targetQueueBiz.list(//
				catalog, key, orderBy, pageNo, pageSize);
		List<TargetQueueEntity> qeLt = pr.getList();
		for (TargetQueueEntity qe : qeLt) {
			qeMap.put(qe.getId(), qe);
			// 将采集到的队列信息插入数据库，如果同名队列已存在则忽略？
			Object ro = brokerAdapter.getMessageBroker(qe.getBroker()).statTargetQueue();
			Map<String, Object> rqMap = (Map<String, Object>) ro;
			for (Entry<String, Object> en : rqMap.entrySet()) {
				String addr = en.getKey();
				ro = en.getValue();
				if (ro == null) {
					log.error("Null result from {}", addr);
					continue;
				}
				if (ro instanceof Throwable) {
					log.error("Remote server {} throw exception", addr, ro);
					continue;
				}
				if (!(ro instanceof Map)) {
					log.error("Remote server {} return unknow object {}", addr, ro);
					continue;
				}
			}
			// 通过查询出监控的状态，改变列表
			Vector<String> tm = targetQueueMonitor.getMoningSet();
			for (TargetQueueEntity b : qeMap.values()) {
				if (b.getStatus() == TargetQueueInfo.STATUS_ENABLE && tm.contains(b.getId())) {
					b.setStatus(TargetQueueInfo.STATUS_MONING);
				}
			}
		}
		qeLt.clear();
		qeLt.addAll(qeMap.values());
		PagedResult<TargetQueueEntity> tqr = new PagedResult<TargetQueueEntity>(//
				qeLt.size(), qeLt, pageNo, pageSize);
		return new ReplyInfo<Object>(true, tqr);
	}

	/**
	 * 修改时，获取表单数据
	 */
	@RequestMapping("/get.do")
	@Authorized(code = "message.target.queue.get")
	public ReplyInfo<Object> get(String id) {
		TargetQueueEntity pr = targetQueueBiz.get(id);
		return new ReplyInfo<Object>(true, pr);
	}

	/**
	 * 新增或修改队列，将消息队列配置保存到数据库
	 */
	@RequestMapping("/save.do")
	@Authorized(code = "message.target.queue.save")
	public ReplyInfo<Object> save(TargetQueueEntity o) {
		if (Util.strIsEmpty(o.getId())) {
			log.warn("Create target queue {}", o.getId());
			o.setStatus(TargetQueueInfo.STATUS_ENABLE);
			// broker.addTargetQueue(o.getId(), o.getXml());
		} else {
			log.warn("Update target queue {}", o.getId());
			// o.setStatus(TargetQueueInfo.STATUS_ENABLE);
			// broker.updateTargetQueue(o.getId(), o.getXml());
		}
		targetQueueBiz.save(o);
		// 上传配置中心

		return ReplyInfo.SUCCESS;
	}

	/**
	 * 从数据库删除队列，并同步配置到所有broker
	 */
	@RequestMapping("/delete.do")
	@Authorized(code = "message.target.queue.delete")
	public ReplyInfo<Object> delete(String id) {
		String broker =targetQueueBiz.get(id).getBroker();
		TargetQueueEntity o = targetQueueBiz.delete(id);
		brokerAdapter.getMessageBroker(broker).removeTargetQueue(o.getId(), false);
		return ReplyInfo.SUCCESS;
	}

	/**
	 * 开始监控
	 */
	@RequestMapping("/start.do")
	@Authorized(code = "message.target.queue.start")
	public ReplyInfo<Object> start(String id, String code) {
		Vector<String> tm = targetQueueMonitor.getMoningSet();
		TargetQueueEntity o = targetQueueBiz.get(id);
		if (!tm.contains(code)) {
			tm.add(code);
		}
		o.setStatus(TargetQueueInfo.STATUS_MONING);
		targetQueueBiz.save(o);
		return ReplyInfo.SUCCESS;
	}

	/**
	 * 停止监控
	 */
	@RequestMapping("/stop.do")
	@Authorized(code = "message.target.queue.stop")
	public ReplyInfo<Object> stop(String id, String code) {
		Vector<String> tm = targetQueueMonitor.getMoningSet();
		TargetQueueEntity o = targetQueueBiz.get(id);
		tm.remove(code);
		o.setStatus(TargetQueueInfo.STATUS_ENABLE);
		return ReplyInfo.SUCCESS;
	}

	/**
	 * 启用队列
	 */
	@SuppressWarnings("unchecked")
	@RequestMapping("/enable.do")
	@Authorized(code = "message.target.queue.enable")
	public ReplyInfo<Object> enable(String id) {
		// 查询队列信息
		TargetQueueEntity o = targetQueueBiz.get(id);
		if (o == null) {
			return Constants.REPLY_ERROR_NOTFOUND;
		}
		// 广播调用message-server，激活队列
		Map<String, Object> rm = null;
		try {
			String broker =targetQueueBiz.get(id).getBroker();
			rm = (Map<String, Object>) brokerAdapter.getMessageBroker(broker).activeTargetQueue(o.getCode());
		} catch (Throwable e) {
			log.error("Error on active target queue {}", o.getCode(), e);
			return Constants.REPLY_ERROR_INTERNAL;
		}
		if (rm == null || rm.isEmpty()) {
			log.warn("No message server to active target queue {}", o.getCode());
			return ReplyInfo.SUCCESS;
		}
		boolean flag = true;
		for (String k : rm.keySet()) {
			ReplyInfo<Object> r = (ReplyInfo<Object>) rm.get(k);
			flag = flag && r.isSuccess();
			if (r.isSuccess()) {
				log.info("Active target queue {} successfully", o.getCode());
			} else {
				log.error("Active target queue {} failed, {}", o.getCode(), r.getData());
			}
		}
		if (flag) {
			o.setStatus(TargetQueueInfo.STATUS_ENABLE);
			targetQueueBiz.save(o);
			return ReplyInfo.SUCCESS;
		}
		return new ReplyInfo<Object>(rm);
	}

	/**
	 * 停用队列
	 */
	@SuppressWarnings("unchecked")
	@RequestMapping("/disable.do")
	@Authorized(code = "message.target.queue.disable")
	public ReplyInfo<Object> disable(String id) {
		// 查询队列信息
		TargetQueueEntity o = targetQueueBiz.get(id);
		if (o == null) {
			return Constants.REPLY_ERROR_NOTFOUND;
		}
		// 广播调用message-server，激活队列
		Map<String, Object> rm = null;
		try {
			String broker =targetQueueBiz.get(id).getBroker();
			rm = (Map<String, Object>) brokerAdapter.getMessageBroker(broker).deactiveTargetQueue(o.getCode());
		} catch (Throwable e) {
			log.error("Error on deactive target queue {}", o.getCode(), e);
			return Constants.REPLY_ERROR_INTERNAL;
		}
		if (rm == null || rm.isEmpty()) {
			log.warn("No message server to deactive target queue {}", o.getCode());
			return ReplyInfo.SUCCESS;
		}
		boolean flag = true;
		for (String k : rm.keySet()) {
			ReplyInfo<Object> r = (ReplyInfo<Object>) rm.get(k);
			flag = flag && r.isSuccess();
			if (r.isSuccess()) {
				log.info("Deactive target queue {} successfully", o.getCode());
			} else {
				log.error("Deactive target queue {} failed, {}", o.getCode(), r.getData());
			}
		}
		if (flag) {
			o.setStatus(TargetQueueInfo.STATUS_DISABLE);
			targetQueueBiz.save(o);
			return ReplyInfo.SUCCESS;
		}
		return new ReplyInfo<Object>(rm);
	}

	/**
	 * 删除队列
	 */
	@RequestMapping("/remove.do")
	@Authorized(code = "message.target.queue.remove")
	public ReplyInfo<Object> remove(String id, boolean force) {
		log.warn("Remove target queue {}", id);
		TargetQueueEntity o = targetQueueBiz.get(id);
		targetQueueBiz.delete(id);
		brokerAdapter.getMessageBroker(o.getBroker()).removeTargetQueue(o.getId(), force);
		return ReplyInfo.SUCCESS;
	}
}